package com.lagou.mall.esflow.server.handler;

import static com.lagou.mall.esflow.server.constant.ESFlowConstant.ThreadConfiguration.TIME_THREAD_LOCAL;
import static com.lagou.mall.esflow.server.constant.ESFlowConstant.ThreadConfiguration.TRACE_THREAD_LOCAL;

import com.lagou.mall.esflow.client.AckEnum;
import com.lagou.mall.esflow.client.Binlog;
import com.lagou.mall.esflow.server.util.FilterEnvUtils;
import com.lagou.mall.esflow.server.util.KafkaMsgUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;


/**
 * @Author xiazhenyu
 * @Description 消息处理Runnable
 * @Date 5:22 下午 2021/2/24
 **/
@Slf4j
public class MessageRunnable implements Runnable {

    private KafkaConsumer<String, String> consumer;
    private ConsumerRecord<String, String> message;
    private AbstractTopicConsumerHandler topicConsumerHandler;


    public MessageRunnable(KafkaConsumer<String, String> consumer, ConsumerRecord<String, String> message,
            AbstractTopicConsumerHandler topicConsumerHandler) {
        this.consumer = consumer;
        this.message = message;
        this.topicConsumerHandler = topicConsumerHandler;
    }

    @Override
    public void run() {
        Binlog binlog = KafkaMsgUtil.parseBinlogData(message.value());
        long startTime = System.currentTimeMillis();
        String id = topicConsumerHandler.getId(binlog);
        String traceId = consumer + "_" + id + "_" + message.key();
        AckEnum ackEnum = AckEnum.UNACK;
        Throwable throwable = null;
        try {
            TRACE_THREAD_LOCAL.set(traceId);
            log.info("msgBeforeExecute msgId:{},table:{},id:{},type:{},publishTime:{}", message.key(),
                    binlog.getTable(), id, binlog.getType(), message.timestamp());
            if (log.isDebugEnabled()) {
                log.debug(binlog.toString());
            }
            if (FilterEnvUtils.filter(binlog)) {
                ackEnum = AckEnum.ACK;
            } else {
                ackEnum = topicConsumerHandler.start(consumer, binlog);
            }
            //TODO 判断是提交
        } catch (Throwable e) {
            throwable = e;
            log.error("topicConsumerHandler process topic fail,handler ", e);
            ackEnum = AckEnum.UNACK;
        } finally {
            //TODO retry 机制
            long time = System.currentTimeMillis() - startTime;
            TIME_THREAD_LOCAL.set(time);
            log.info("msgAfterExecute totalTime: {}, mode_{}", time, ackEnum.name());
            TIME_THREAD_LOCAL.remove();
        }
    }

}
